package com.zysd.collection.scheduler.service.amqp.impl;

import com.lensyn.collect.common.enums.ProtocolEnum;
import com.lensyn.collect.meters.codec.applayer.SequenceDomain;
import com.lensyn.collect.meters.codec.linklayer.*;
import com.lensyn.collect.meters.netty.MessageHelper;
import com.lensyn.collect.meters.store.MetersMessage;
import com.lensyn.common.exception.entity.MsgException;
import com.zysd.collection.scheduler.entity.Operate;
import com.zysd.collection.scheduler.entity.Terminal;
import com.zysd.collection.scheduler.service.amqp.AmqpClientService;
import com.zysd.collection.scheduler.service.operate.OperateService;
import com.zysd.collection.scheduler.service.record.RecordApiService;
import com.zysd.collection.scheduler.system.config.SysConfig;
import com.zysd.collection.scheduler.system.util.EncodeUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 消息mq推送服务类
 *
 * @author hetao
 */
@Service
public class AmqpClientServiceImpl implements AmqpClientService {

    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final Logger logger = LoggerFactory.getLogger(AmqpClientServiceImpl.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${collectd.meters.rabbitmq.send-exchange}")
    private String sendExchange;

    @Value("${collectd.meters.rabbitmq.send-routingKey}")
    private String sendRoutingKeyPrefix;

    @Autowired
    private OperateService operateService;

    @Autowired
    private RecordApiService apiService;

    /**
     * rabbit异步发送消息
     * 需要指定交换机和routingKey
     * rabbitTemplate.sendAndReceive(); 同步发送消息并接收返回结果
     *
     * @param terminal 终端信息
     * @param msg      发送消息对象
     */
    @SuppressWarnings("unchecked")
    private void send(Terminal terminal, MetersMessage msg) {
        //终端规约版本信息维护
        ProtocolEnum protocolEnum = SysConfig.convert(terminal.getRegionNo(), terminal.getLocalNo());
        if (protocolEnum == null) {
            Terminal terminal1 = apiService.getCusTerminalById(terminal.getId());
            SysConfig.set(terminal1.getRegionNo(), terminal1.getLocalNo(), terminal1.getProtocol());
        }
        try {
            msg.region = terminal.getRegionNo();
            msg.local = terminal.getLocalNo();
            String routingKey = sendRoutingKeyPrefix + "." + terminal.getIpAddr() + "." + terminal.getIpPort();
            Operate operate = saveOperate(terminal, msg);
            if (msg.tsuid == null) {
                msg.tsuid = String.valueOf(operate.getId());
            }
            logger.info("成功发送消息: 发送消息队列={}；发送消息路由(routingKey)={}; 发送消息内容={}", sendExchange, routingKey, msg);
            rabbitTemplate.convertAndSend(sendExchange, routingKey, msg);
        } catch (Exception e) {
            logger.error("消息推送失败", e);
            throw new MsgException("消息推送失败", e);
        }
    }


    /**
     * 推送msg消息
     */
    @Override
    public void publish(Terminal terminal, MetersMessage msg) {
        send(terminal, msg);
    }

    /**
     * 保存操作日志
     */
    @Transactional(rollbackFor = Exception.class)
    public Operate saveOperate(Terminal terminal, MetersMessage msgs) {
        msgs.region = terminal.getRegionNo();
        msgs.local = terminal.getLocalNo();
        UserDataDomain udata = MessageHelper.wrapMessage(PooledByteBufAllocator.DEFAULT, msgs);
        udata.ctrl.dir = Direction.ToTerminal;
        udata.ctrl.prm = Primary.Primary;
        udata.ctrl.code = Function.getFunctionCode(udata.ctrl.prm, udata.data().afn);
        udata.address.region = msgs.region;
        udata.address.terminal = msgs.local;
        int bytes = 1 + 5 + 1 + 1;
        bytes += udata.data().content().readableBytes();
        byte[] array = new byte[bytes];
        ByteBuf cbuf = Unpooled.wrappedBuffer(array);
        cbuf.resetWriterIndex();
        ControlDomain.Pipe.INSTANCE.put(cbuf, udata.ctrl);
        AddressDomain.Pipe.INSTANCE.put(cbuf, udata.address);
        cbuf.writeByte(udata.data().afn.value);
        SequenceDomain.Pipe.INSTANCE.put(cbuf, udata.data().seq);
        cbuf.writeBytes(udata.data().content());

        Operate operate = new Operate();
        operate.setOccur(sdf.format(new Date()));
        operate.setRegion(terminal.getRegionNo());
        operate.setLocal(terminal.getLocalNo());
        operate.setAfn(msgs.afn);
        operate.setReq(EncodeUtil.hex(array));
        operateService.insert(operate);
        return operate;
    }
}
